package bootsxt.websocket;

import bootsxt.utils.Kafka;
import bootsxt.utils.MsgType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yeauty.annotation.*;
import org.yeauty.pojo.Session;

import java.io.IOException;

/**
 * WebSocket基类，管理任务的启动和停止。
 * <p>
 * 主要改动：
 * 1. 增加subscriptionId成员变量，方便统一退订；
 * 2. 在 onOpen / onClose / onError 中都能正常退订；
 * 3. 优化日志与异常处理，减少 throw RuntimeException；
 * 4. 用更语义化的方法抽取“构建订阅ID”和“退订”逻辑，避免重复拼接字符串。
 */
public abstract class WebSocketBase {

    protected final Logger log = LoggerFactory.getLogger(this.getClass());

    /** 当前会话（客户端连接） */
    private Session session;
    /** 当前用户“本地”可生产消息的Topic */
    private Kafka.Topic<byte[]> localTopic;
    /** 用于订阅对方Topic时的订阅ID */
    private String subscriptionId;

    /**
     * 子类必须实现，返回本WebSocket是用于 AUDIO 还是 IMAGE（或其他类型）
     */
    protected abstract MsgType getMsgType();

    /**
     * 当客户端向服务端发送二进制数据时触发
     * - 这里简单做：收到消息后，若本地Topic有订阅者就produce消息
     */
    @OnBinary
    public void onBinary(Session session,
                         @RequestParam String id,
                         byte[] bytes) throws IOException {
        // 如果当前topic还有订阅者，就把二进制数据广播出去
        if (localTopic != null && !localTopic.getScribes().isEmpty()) {
            localTopic.produce(bytes);
        }
    }

    /**
     * 当客户端连接建立时触发
     * - 创建/获取“本地”topic，用于生产消息
     * - 去对方Topic上订阅数据，收到对方消息后发送给当前session
     * - 启动一个线程，定期打印订阅信息（可选）
     */
    @OnOpen
    public void onOpen(Session session,
                       @RequestParam String id,
                       @RequestParam String toId) {
        this.session = session;
        // 构造本地topic：发送方 (id)
        this.localTopic = Kafka.topic(id, getMsgType().name(), byte[].class);
        // 构造订阅ID，用于订阅对方Topic
        this.subscriptionId = buildSubscriptionId(id, toId, session);

        log.info("正在连接: {} => 目标ID: {}", id, toId);

        // 向对方Topic订阅
        Kafka.topic(toId, getMsgType().name(), byte[].class)
                .scribes(bytes -> {
                    try {
                        // 有数据就推送给当前WebSocket客户端
                        session.sendBinary(bytes);
                    } catch (Throwable e) {
                        log.error("发送出错, 准备退订: {}", subscriptionId, e);
                        // 发送失败说明客户端断开等异常，执行退订
                        unsubscribeTopic(toId, subscriptionId);
                    }
                    return null;
                }, subscriptionId);

        log.info("已连接: {} => 目标ID: {}", id, toId);

        // 启动一个线程，仅用来演示“定时打印订阅状态”
        new Thread(() -> {
            while (session.isOpen()) {
                log.info("id:{} 订阅:{} {}，订阅我的: {}",
                        id,
                        toId,
                        getMsgType().name(),
                        localTopic.getScribes().keySet().stream().toList());
                try {
                    Thread.sleep(10_000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }, "MonitorThread-" + session.id()).start();
    }

    /**
     * 当客户端关闭连接（或异常断开）时触发
     * - 做退订操作，防止“僵尸订阅”
     */
    @OnClose
    public void onClose(Session session,
                        @RequestParam String id,
                        @RequestParam String toId) {
        log.info("即将 onClose: {} => 目标ID: {}", id, toId);
        try {
            // 退订对方Topic
            unsubscribeTopic(toId, subscriptionId);
        } finally {
            if (session.isOpen()) {
                session.close();
            }
        }
        log.info("完成 onClose: {} => 目标ID: {}", id, toId);
    }

    /**
     * 当发生异常时触发
     * - 这里同样退订
     */
    @OnError
    public void onError(Session session,
                        @RequestParam String id,
                        @RequestParam String toId,
                        Throwable throwable) {
        log.error("{}: 报错了 session:{}, 原因: {}",
                getMsgType().name(),
                session.id().toString(),
                throwable.getMessage(),
                throwable);

        unsubscribeTopic(toId, subscriptionId);
    }

    /**
     * 若外部需要直接拿本地Topic，可提供getter
     */
    public Kafka.Topic<byte[]> getTopic() {
        return this.localTopic;
    }

    /**
     * 构建订阅ID，方便在多个地方退订
     */
    private String buildSubscriptionId(String id, String toId, Session session) {
        return "websocket: "+id
                + ">>>" + toId
                + "_"   + session.remoteAddress()
                + "_"   + session.id();
    }

    /**
     * 根据 toId + subscriptionId 退订对应的topic
     */
    private void unsubscribeTopic(String toId, String subscriptionId) {
        if (subscriptionId == null || toId == null) {
            return;
        }
        Kafka.topic(toId, getMsgType().name(), byte[].class).unsubscribe(subscriptionId);
    }

}
